/**
 * 
 */
package edu.vanderbilt.vshare.multicast;

import java.net.DatagramPacket;
import java.net.InetAddress;
import java.net.MulticastSocket;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentLinkedQueue;


/**
 * thread for sending file data across a multicast ip
 * 
 * @author jedmondson
 *
 */
public class MulticastSenderThread implements Runnable {
	
	public MulticastSenderThread()
	{		
		failed = true;
	}

	/**
	 * Constructor for a multicast sending thread
	 * 
	 * @param    address     multicast address to talk on
	 * @param    fileid      file id that this thread will broadcast
	 * @param    rate        rate to service a work unit
	 *                       specified as 256KBs, 16KBs, etc.
	 */
	
	public MulticastSenderThread(byte [] address, int fileid, String rate) {
		
		setAddress (address);
		setFileid (fileid);

		//setRate (rate);
		this.rate = rate;
	}

	/**
	 * Constructor for a multicast sending thread
	 * 
	 * @param    host        multicast address to talk on
	 * @param    fileid      file id that this thread will broadcast
	 * @param    rate        rate to service a work unit
	 *                       specified as 256KBs, 16KBs, etc.
	 */
	
	public MulticastSenderThread(String host, int fileid, String rate) {
		
		setAddress (host);
		setFileid (fileid);

		//setRate (rate);
		this.rate = rate;
	}

	/**
	 * reset the failed flag. Useful if reusing a previous thread that had failed.
	 */
	
	public void resetFailed ()
	{
		failed = false;
	}

	/**
	 * check to see if this thread failed in its initialization methods
	 * @return                   true if thread can be run
	 */
	
	public boolean isFailed ()
	{
		return failed;
	}

	/**
	 * set the multicast address that this thread will send on
	 * @param     address        multicast ip to send to
	 * @param     port           port to bind to
	 * @return                   true if thread can be run
	 */
	
	public boolean setAddress (byte [] address, int port)
	{
		try {
			addr = InetAddress.getByAddress(address);
			socket = new MulticastSocket(port);
			socket.joinGroup(addr);		
			this.port = port;
		}
		catch (Exception e)
		{
			failed = true;
		}
		
		return failed;
	}

	/**
	 * set the multicast address that this thread will send on
	 * @param     host           multicast ip to send to
	 * @param     port           port to bind to
	 * @return                   true if thread can be run
	 */
	
	public boolean setAddress (String host, int port)
	{
		try
		{
			InetAddress my_addr = InetAddress.getByName(host);
			return setAddress (my_addr.getAddress(), port);
		}
		catch (Exception e)
		{
			failed = true;
		}
		
		return failed;
	}

	/**
	 * set the file that we will be reading from and sending the data of
	 * @param     fileid         file to retrieve from the global store
	 * @return                   true if successful
	 */
	
	public boolean setFileStruct (int fileid)
	{
		if (SharedGlobals.store.contains(fileid))
			fileStruct = SharedGlobals.store.get(fileid);
		
		if (fileStruct != null)
		{
			createWeightedWorkUnits();
			println ("Setting up packet list with " + fileStruct.getPackets() + 
					" packets for file " + fileStruct.getName ());
			return true;
		}
		
		println ("Failed to read file " + fileid);
		
		return failed;
	}

	/**
	 * created a weighted work array, which is currently unused but will be used
	 * once again when priority queues are reimplemented
	 */
	
	private void createWeightedWorkUnits ()
	{
		int numPackets = getPackets (packetSize);
		
		weights = new WeightedWorkUnit[numPackets + 1];
		
		for (int i = -1; i < numPackets; ++i)
		{
			weights[i+1] = new WeightedWorkUnit(null,fileid,i,0);
		}
	}

	/**
	 * set the multicast address that this thread will send on
	 * @param     address           multicast ip to send to
	 */
	
	public boolean setAddress (byte [] address)
	{
		return setAddress (address, this.port);
	}

	/**
	 * set the multicast address that this thread will send on
	 * @param     host               multicast ip to send to
	 */
	
	public boolean setAddress (String host)
	{
		return setAddress (host, this.port);
	}

	/**
	 * get this file's meta data. If, for some reason, the file
	 * had no meta data, then this function will create it. The
	 * meta data returned is NOT padded to 16 KB. This is meant
	 * to be appended onto a server repo metadata block.
	 * @return                     the file meta data
	 */
	
	public byte [] getMeta ()
	{
		if (this.meta == null)
		{
			StringBuilder meta = new StringBuilder ();
			/*
			meta.append("ip: ");
			meta.append(addr.getHostAddress());
			meta.append("\nport: ");
			meta.append(port);
			meta.append("\nrate: ");
			meta.append(rate);
			meta.append("\nfile: ");
			meta.append(fileid);
			meta.append("\n");
			*/
			
			meta.append ("  <address>\n");
			meta.append ("    <ip>" + addr.getHostAddress() + "</ip>\n");
			meta.append ("    <port>" + port + "</port>\n");
			meta.append ("    <rate>" + rate + "</rate>\n");
			meta.append ("    <fileid>" + fileid + "</fileid>\n");
			meta.append ("  </address>\n");
			
			this.meta = meta.toString().getBytes ();
		}
		return this.meta;
	}

	/**
	 * set the data rate to publish at. 
	 * @param     rate               data rate to publish at. Can be of the following form:
	 *                               (Number)KBs, e.g. 16KBs for 16 Kilobytes per second
	 * @return                       true if successful
	 */
	
	public boolean setRate (String rate)
	{		
		this.rate = rate;
		int end = rate.indexOf("KBs");
	
		if (end != -1)
		{
			int data = Integer.parseInt(rate.substring(0, end));
			int packets = data / (packetSize / 1024);			
			int wakeup = 1000 / packets;
			
			if (!failed)
				timer.scheduleAtFixedRate(new TimerTask () {
					public void run ()
					{
						// timer task execution
						// if the server is not shutdown
						if (!SharedGlobals.shutdown.get())
						{
							try {
								// remove a work unit
								Integer cur = remove ();
								if (cur != null)
								{
									// send the packet out
									println ("Packet " + cur + " request.");
									
									DataPacket checker = new DataPacket (fileid, cur, 
											MulticastSenderThread.this.fileStruct.getPayload(cur));
									
									byte[] message = checker.toBytes();
									
									//DatagramPacket packet = new DatagramPacket(message,
									//		message.length);
									DatagramPacket packet = new DatagramPacket(message, 
											message.length, addr, port);
									
									socket.send (packet);

									println ("Packet " + cur + " of length "
											+ message.length + " was sent.");
									
								}
							} 
							catch (Exception e) {
								e.printStackTrace();
							}
						} // end if !shutdown
						else
						{
							// if shutdown, cancel the task
							this.cancel();
						}
						
					}
				}, 0, wakeup);
		}
		
		else
			failed = true;
		
		return failed;
	}
	
	/**
	 * @return the fileid
	 */
	public int getFileid() {
		return fileid;
	}

	/**
	 * @param fileid the fileid to set
	 */
	public void setFileid(int fileid) {
		this.fileid = fileid;
		
		setFileStruct (fileid);
	}

	/**
	 * start the timertask
	 */
	
	public void run() {

		if (!failed)
		{
			setRate (rate);
		}
		
		
		//System.out.println ("Executed...");
	}

	/**
	 * add a work unit to this thread. Without calling this method, the thread will
	 * loop and accomplish no real work.
	 * @param     packet_start       packet index to send
	 * @param     number             number of packets to send (packet_start inclusive)
	 */
	
	public void add (int packet_start, int number)
	{
		for (int i = 0; i < number && packet_start + i < fileStruct.getPackets(); ++i)
			add (packet_start + i);
	}

	/**
	 * add a work unit to this thread. Without calling this method, the thread will
	 * loop and accomplish no real work.
	 * @param     packet             packet index to send
	 */
	
	public synchronized void add (Integer packet)
	{
		// if weight == 0, then we need to add it to the work queue
		if (!queue.contains(packet))
			queue.add(packet);
	}

	/**
	 * formatted print line method that will indicate the thread by
	 * the multicast ip it is sending to
	 * 
	 * @param     message             message to print
	 */
	
	public synchronized void println (String message)
	{
		System.err.println(addr.getHostAddress() + ": " +
				message);
	}

	/**
	 * returns the number of packets that a file will be divided into
	 * 
	 * @param     fileid             id of file to look at
	 */
	
	public int getPackets (int fileid)
	{
		return fileStruct.getPackets();
	}

	/**
	 * removes a work unit from the queue
	 * 
	 * @return                       unit that was removed        
	 */
	
	public synchronized Integer remove ()
	{	
		return queue.poll();
	}
	
	private Timer timer = new Timer ();
	private boolean failed = false;
	private int fileid;
	//private PriorityQueue<WeightedWorkUnit> queue = new PriorityQueue<WeightedWorkUnit> 
    //               (100, new WeightedWorkUnitComparator ());;

    private ConcurrentLinkedQueue <Integer> queue = new ConcurrentLinkedQueue <Integer> ();
             
    //private ConcurrentHashMap <Integer, WeightedWorkUnit> map = 
    //	new  ConcurrentHashMap <Integer, WeightedWorkUnit> ();

    WeightedWorkUnit[] weights = null;
	
	private MulticastSocket socket; 
	private InetAddress addr;
	private int packetSize = 16384;
	private int port = 10000;
	private FileStruct fileStruct = null;
	private String rate = null;
	
	byte [] meta;
}
